package com.kawins.supers.message.socket.manager;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.kawins.base.share.BaseConstant;
import com.kawins.supers.message.socket.IBeforeSend;
import com.kawins.supers.message.socket.IWebSocket;
import com.kawins.supers.message.socket.bean.SocketMessage;
import com.kawins.supers.message.socket.listener.MessageBean;
import com.kawins.supers.message.socket.listener.SocketMessageListener;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.io.Serializable;
import java.util.List;
import java.util.Set;

/**
 * @author rankai
 *         createTime 2018-01-2018/1/22 17:27
 */
public class WebSocketImpl extends SocketMessageListener implements IWebSocket {

    private static final Logger log = LoggerFactory.getLogger(WebSocketImpl.class);

    //redis_Session消息默认前缀
    private static final String DEF_WEB_SOCKET_REDIS_MSG_KEY = "webSocket.message";
    //redis_Session消息前缀
    private String webSocketRedisMsgKey = DEF_WEB_SOCKET_REDIS_MSG_KEY;
    //redis_Session消息发出位置标识,标识是那一台服务器发出的
    private String webSocketServerName = SOCKET_SERVER_UUID;
    //是否发送消息
    private boolean sendRedisMsg = true;

    private RedisTemplate<String, Object> redisTemplate;
    //IBeforeSend接口
    private IBeforeSend beforeSend;

    @Override
    public void sendMessage(Serializable key, SocketMessage message, boolean online) {
        if (key != null && message != null) {
            List<WebSocketSession> sessions = readSession(key, message);
            if (sessions != null) {
                for (WebSocketSession session : sessions) {
                    send(key, session, message, true);
                }
            } else {
                if (!online) {
                    send(key, null, message, false);
                }
            }
        }
    }

    @Override
    public void sendMessage(Serializable key, Serializable sessionId, SocketMessage message) {
        if (key != null && message != null) {
            WebSocketSession webSocketSession = readSession(key, sessionId, message);
            if (webSocketSession != null) {
                send(key, webSocketSession, message, true);
            }
        }
    }

    @Override
    public void sendMessage(Serializable key, List<SocketMessage> messages, boolean online) {
        if (key != null && messages != null) {
            for (SocketMessage message : messages) {
                sendMessage(key, message, online);
            }
        }
    }

    @Override
    public void sendMessage(Serializable key, Serializable sessionId, List<SocketMessage> messages) {
        if (key != null && sessionId != null && messages != null) {
            for (SocketMessage message : messages) {
                sendMessage(key, sessionId, message);
            }
        }
    }

    @Override
    public void sendMessage(Set<Serializable> keys, SocketMessage message, boolean online) {
        if (keys != null && message != null) {
            for (Serializable key : keys) {
                sendMessage(key, message, online);
            }
        }
    }

    @Override
    public void sendMessage(SocketMessage socketMessage) {
        if (socketMessage != null) {
            sendRedisMsg(3, null, null, socketMessage);
            sendAll(socketMessage);
        }
    }

    @Override
    protected void onMessage(MessageBean message) {
        String serverName = message.msgBody.serverName;
        if (StringUtils.isBlank(serverName) || serverName.equals(webSocketServerName)) {
            return;
        }
        Integer type = message.msgBody.type;
        SocketMessage socketMessage = message.msgBody.socketMessage;
        log.debug("receive broadcast, broadcast type is {} ", type);
        switch (type) {
            case 1: {
                Serializable key = message.msgBody.key;
                Serializable sessionId = message.msgBody.sessionId;
                if (key != null && sessionId != null) {
                    WebSocketSession webSocketSession = WebSocketBank.get(key, sessionId);
                    if (webSocketSession != null) {
                        send(key, webSocketSession, socketMessage, true);
                    }
                } else {
                    log.warn("broadcast type is {}, but key or sessionId is null", type);
                }
                break;
            }
            case 2: {
                Serializable key = message.msgBody.key;
                if (key != null) {
                    List<WebSocketSession> sessions = WebSocketBank.get(key);
                    if (sessions != null) {
                        for (WebSocketSession session : sessions) {
                            send(key, session, socketMessage, true);
                        }
                    }
                } else {
                    log.warn("broadcast type is {}, but key is null", type);
                }
                break;
            }
            case 3: {
                sendAll(socketMessage);
            }
            default: {
                log.warn("unknown receive type {} ", type);
                break;
            }
        }
    }

    /**
     * 读取WebSocketSession
     *
     * @param buildKey
     * @param socketMessage
     * @return
     */
    protected List<WebSocketSession> readSession(Serializable buildKey, SocketMessage socketMessage) {
        List<WebSocketSession> webSocketSessions = WebSocketBank.get(buildKey);
        sendRedisMsg(2, buildKey, null, socketMessage);
        return webSocketSessions;
    }

    /**
     * 读取WebSocketSession
     *
     * @param buildKey
     * @param sessionId
     * @param socketMessage
     * @return
     */
    protected WebSocketSession readSession(Serializable buildKey, Serializable sessionId, SocketMessage socketMessage) {
        WebSocketSession webSocketSession = WebSocketBank.get(buildKey, sessionId);
        if (webSocketSession != null) {
            return webSocketSession;
        } else {
            sendRedisMsg(1, buildKey, sessionId, socketMessage);
        }
        return null;
    }


    /**
     * 发送全部消息
     *
     * @param message
     * @param message
     */
    protected void sendAll(SocketMessage message) {
        List<WebSocketSession> webSocketSessions = WebSocketBank.getAll();
        if (webSocketSessions != null) {
            for (WebSocketSession webSocketSession : webSocketSessions) {
                send(null, webSocketSession, message, false);
            }
        }
    }

    /**
     * 发送消息
     *
     * @param buildKey      发送人key
     * @param socketSession WebSocketSession
     * @param message       消息内容
     * @param online        是否只发在线
     */
    private void send(Serializable buildKey, WebSocketSession socketSession, SocketMessage message, boolean online) {
        if (message != null) {
            boolean isSend = false;
            JSONObject toJSON = (JSONObject) JSON.toJSON(message);
            if (online || buildKey == null) {
                if (socketSession != null) {
                    isSend = true;
                    if (beforeSend != null && message.getId() == null) {
                        Object msgId = beforeSend.beforeSend(socketSession.getAttributes()
                                .get(BaseConstant.USER_ID_SESSION_KEY), message);
                        toJSON.put("id", msgId);
                    }
                }
            } else {
                if (beforeSend != null && message.getId() == null) {
                    Object msgId = beforeSend.beforeSend(buildKey, message);
                    toJSON.put("id", msgId);
                }
                if (socketSession != null) {
                    isSend = true;
                }
            }
            if (isSend && socketSession != null && toJSON != null) {
                try {
                    socketSession.sendMessage(new TextMessage(toJSON.toJSONString()));
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private void sendRedisMsg(Integer type, Serializable buildKey, Serializable sessionId, SocketMessage socketMessage) {
        if (sendRedisMsg && redisTemplate != null) {
            log.debug("not find {} WebSocketSession, start broadcast", buildKey);
            MessageBean.MessageBody messageBody = new MessageBean
                    .MessageBody(webSocketServerName, type, buildKey, sessionId, socketMessage);
            redisTemplate.convertAndSend(webSocketRedisMsgKey, messageBody);
        }
    }

    public void setWebSocketRedisMsgKey(String webSocketRedisMsgKey) {
        this.webSocketRedisMsgKey = webSocketRedisMsgKey;
    }

    public String getWebSocketRedisMsgKey() {
        return webSocketRedisMsgKey;
    }

    public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public void setBeforeSend(IBeforeSend beforeSend) {
        this.beforeSend = beforeSend;
    }

    public void setSendRedisMsg(boolean sendRedisMsg) {
        this.sendRedisMsg = sendRedisMsg;
    }

    public void setWebSocketServerName(String webSocketServerName) {
        this.webSocketServerName = webSocketServerName;
    }
}
